package mq

import (
	"gim/comm"
	"bytes"
	"github.com/garyburd/redigo/redis"
	log "github.com/golang/glog"
)

type IMQ interface {
	Enqueue(queue string, msg *comm.Message) bool
	Dequeue(queue string) *comm.Message
	BDequeue(queue string) *comm.Message
}

type RedisMQ struct {
	redis_pool *redis.Pool
}

func (rmq *RedisMQ) Enqueue(queue string, msg *comm.Message) bool {
	conn := rmq.redis_pool.Get()
	defer conn.Close()
	
	body := msg.ToData()
	buffer := new(bytes.Buffer)
	comm.WriteHeader(int32(len(body)), int64(msg.Seq), byte(msg.Cmd), byte(msg.Version), buffer)
	buffer.Write(body)
	buf := buffer.Bytes()
	
	_, err := conn.Do("RPUSH", queue, buf)
	if err != nil {
		log.Error("redis RPUSH err: redismq.enqueue", err)
		return false
	}
	
	return true
}

func (rmq *RedisMQ) Dequeue(queue string) *comm.Message {
	conn := rmq.redis_pool.Get()
	defer conn.Close()
	
	ret, err := redis.Bytes(conn.Do("LPOP", queue))
	if err != nil {
		log.Error("redis LPOP err: redismq.dequeue", err)
		return nil
	}
	log.Infof("dequeue: %v", ret)
	buff := ret
	length, seq, cmd, version := comm.ReadHeader(buff[0:16])
	if length < 0 || length > 64*1024 {
		log.Info("invalid len:", length)
		return nil
	}

	message := new(comm.Message)
	message.Cmd = cmd
	message.Seq = seq
	message.Version = version
	if !message.FromData(buff[16:]) {
		log.Warning("parse error")
		return nil
	}
	return message
}

func (rmq *RedisMQ) BDequeue(queue string) *comm.Message {
	conn := rmq.redis_pool.Get()
	defer conn.Close()
	
	ret, err := redis.Values(conn.Do("BLPOP", queue, 300))
	if err != nil {
		log.Error("redis LPOP err: redis mq.dequeue", err)
		return nil
	}

	if len(ret) != 2 {
		log.Error("redis LPOP err: redis mq.dequeue ret len err")
		return nil
	}
	
	buff := ret[1].([]byte)
	length, seq, cmd, version := comm.ReadHeader(buff[0:16])
	if length < 0 || length > 64*1024 {
		log.Info("invalid len:", length)
		return nil
	}
	log.Infof("length:%d, seq:%d, cmd:%d, version:%d", length, seq, cmd, version)
	message := new(comm.Message)
	message.Cmd = cmd
	message.Seq = seq
	message.Version = version
	if !message.FromData(buff[16:]) {
		log.Warning("parse error")
		return nil
	}
	
	return message
}

func NewRedisMQ(host string, password string) *RedisMQ{
	rmq := new(RedisMQ)
	rmq.redis_pool = comm.NewRedisPool(host, password)
	
	return rmq
}

func StopRedisMQ(rmq *RedisMQ) {
	rmq.redis_pool.Close()
}